package com.pangyangyang.canal.sample;/*
 * Created by gangzz.
 *  2015/7/3
 */

import com.alibaba.otter.canal.client.impl.SimpleCanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 *
 */
public class SimpleClientFailOver {

    private String destination = "example";
    private SimpleCanalConnector connector;

    public static void main(String[] args){
        new SimpleClientFailOver().run();
    }

    public void run(){
        connector = new SimpleCanalConnector(new InetSocketAddress("10.128.170.167",
                11111), "", "", destination);
        boolean retry = true;
        while (retry) {
            retry = false;
            try {
                connector.connect();
                connector.subscribe();
                while (true) {
                    Message message = connector.getWithoutAck(10, 1L, TimeUnit.SECONDS);
                    List<CanalEntry.Entry> entryList = message.getEntries();
                    System.out.println("receive size " + entryList.size() + " >>> 可以拔网线测试了...");
                    sleep(1, TimeUnit.SECONDS);
                    connector.rollback();
                }
            } catch (CanalClientException ce) {
                retry = true;
                System.out.println("broken");
                sleep(1, TimeUnit.SECONDS);
                System.out.println("retry connect...");
            } catch (Throwable t) {
                t.printStackTrace();
            } finally {
                try {
                    connector.unsubscribe();
                } catch (Throwable t){}
                try{
                    connector.disconnect();
                }catch (Throwable t){}
            }
        }
    }

    private void sleep(int time, TimeUnit timeUnit){
        try{
            Thread.sleep(timeUnit.toMillis(time));
        } catch (InterruptedException e){

        }
    }
}
